package com.cbh.amadeus.server.handler;

import com.cbh.amadeus.common.domain.AckModel;
import com.cbh.amadeus.common.enums.MessageType;
import com.cbh.amadeus.common.json.Json;
import com.cbh.amadeus.common.protocol.AmadeusMessage;
import com.cbh.amadeus.common.protocol.AmadeusMessageWrap;
import com.cbh.amadeus.server.core.Publisher;
import com.cbh.amadeus.server.domain.AckResultModel;
import com.cbh.amadeus.server.provider.component.PublisherComponent;
import com.cbh.amadeus.server.provider.tool.AckResultModelTool;
import com.fasterxml.jackson.core.type.TypeReference;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import lombok.extern.slf4j.Slf4j;

/**
 * @author ChenBiHui
 * @date 2025/4/25
 */
@Slf4j
public class AckHandler extends SimpleChannelInboundHandler<AmadeusMessageWrap> {

    private final PublisherComponent component;

    public AckHandler(PublisherComponent component) {
        this.component = component;
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, AmadeusMessageWrap msg) {
        MessageType type = MessageType.find(msg.getType());
        if (type == MessageType.ACK) {
            Publisher.DEFAULT.execute(() -> {
                AmadeusMessage message = msg.getMessage();
                AckModel ackModel = Json.parse(message.getPayload(), new TypeReference<AckModel>() {
                });
                if (ackModel == null) {
                    log.warn(
                            "===== [ACK] Topic [{}] message-id [{}] ack-result is null, can't finish =====",
                            message.getTopic(),
                            message.getMessageId()
                    );
                    return;
                }
                AckResultModelTool modelTool = component.getModelTool();
                AckResultModel model = new AckResultModel();
                model.setId(modelTool.sequence());
                model.setTenant(message.getTenant());
                model.setProvider(message.getProvider());
                model.setTopic(message.getTopic());
                model.setMessageId(message.getMessageId());
                model.setConsumer(ackModel.getConsumer());
                model.setAckTime(ackModel.getAckTime());
                modelTool.finish(model);
                log.info(
                        "===== [ACK] Topic [{}] message-id [{}] with [{}] has finish =====",
                        message.getTopic(),
                        message.getMessageId(),
                        ackModel.getConsumer()
                );
            });
            return;
        }
        ctx.fireChannelRead(msg);
    }
}
